package javaVersion.learn.waterMark;

import javaVersion.pojo.WaterMarkOrder;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @program: myFlink
 * @description: 使用水印, 允许数据延迟2秒
 * @author: WincoMa
 * @create: 2020-07-20 17:00
 **/
public class WaterMarkDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置处理时间为EventTime
        senv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<WaterMarkOrder> orderDataStream = senv.addSource(new RichSourceFunction<WaterMarkOrder>() {
            boolean isRunning = true;

            @Override
            public void run(SourceContext<WaterMarkOrder> ctx) throws Exception {
                Random random = new Random();
                while (isRunning) {
                    WaterMarkOrder markOrder = new WaterMarkOrder(UUID.randomUUID().toString(),
                            random.nextInt(3),
                            random.nextInt(101),
                            new Date().getTime());
                    ctx.collect(markOrder);
                    TimeUnit.SECONDS.sleep(1);
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }
        });

        //设置水印
        orderDataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<WaterMarkOrder>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(WaterMarkOrder element) {
                DateFormat instance = SimpleDateFormat.getInstance();
                String format = instance.format("HH:mm:ss");
                System.out.println("当前水印:" + instance.format(getCurrentWatermark().getTimestamp()));
                System.out.println("当前事件时间:" + instance.format(element.getTimestamp()));
                System.out.println("当前系统时间:" + instance.format(System.currentTimeMillis()));
                return element.getTimestamp();
            }
        });

    }
}
